package com.yzh.luckydraw.application.mq.consumer;

import com.alibaba.fastjson.JSON;
import com.yzh.luckydraw.application.mq.producer.KafkaProducer;
import com.yzh.luckydraw.domain.activity.model.vo.ActivityPartakeRecordVO;
import com.yzh.luckydraw.domain.activity.service.partake.IActivityPartake;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Optional;

/**
 * 抽奖活动领取记录监听消息
 */
@Component
public class ActivityPartakeRecordListener {
    private Logger logger = LoggerFactory.getLogger(ActivityPartakeRecordListener.class);
    @Resource
    private IActivityPartake activityPartake;

    @KafkaListener(topics = KafkaProducer.TOPIC_ACTIVITY_PARTAKE, groupId = "luckydraw")
    public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        Optional<?> message = Optional.ofNullable(record.value());
        //实际场景在消费MQ时，可以通过固定标识判断是否已经消费过
        //1.判断消息是否存在
        if (!message.isPresent()) {
            return;
        }
        //2.处理MQ
        try {
            //1.转化对象
            ActivityPartakeRecordVO activityPartakeRecordVO = JSON.parseObject((String) message.get(), ActivityPartakeRecordVO.class);
            logger.info("消费MQ消息，异步扣减活动库存 msg：{}", message.get());

            //2.更新数据库(实际业务体量较大时，可能会由于MQ消费引起并发，对数据库产生压力，可以将库记录放到缓存中，使用定时任务进行处理缓存和数据库库存同步)
            activityPartake.updateActivityStock(activityPartakeRecordVO);

            //3.消息消费完成
            ack.acknowledge();
        } catch (Exception e) {
            logger.error("消费MQ消息失败(更新库存记录) topic：{} msg：{}", topic, message.get());
            throw e;
        }
    }

}
